package com.hisense.ovcloud.warehouse.service;

import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.core.lang.Pair;
import cn.hutool.json.JSONUtil;
import com.google.common.collect.Lists;
import com.hisense.ovcloud.commons.configs.base.BaseResponse;
import com.hisense.ovcloud.commons.tools.AES256CbcUtil;
import com.hisense.ovcloud.commons.tools.HiAESUtil;
import com.hisense.ovcloud.commons.tools.LocalLFUCacheUtil;
import com.hisense.ovcloud.commons.tools.MurMurHashUtils;
import com.hisense.ovcloud.warehouse.Constants;
import com.hisense.ovcloud.warehouse.amq.AMQMsg;
import com.hisense.ovcloud.warehouse.amq.AMQProducer;
import com.hisense.ovcloud.warehouse.config.ReadShardingTablesConfig;
import com.hisense.ovcloud.warehouse.config.ShardingJDBCExecutor;
import com.hisense.ovcloud.warehouse.config.WarehouseProperties;
import com.hisense.ovcloud.warehouse.dto.common.DeviceInfoInCache;
import com.hisense.ovcloud.warehouse.dto.req.*;
import com.hisense.ovcloud.warehouse.dto.resp.ActivateResp;
import com.hisense.ovcloud.warehouse.dto.resp.AuthBatchResp;
import com.hisense.ovcloud.warehouse.dto.resp.AuthTokenAndBatchResp;
import com.hisense.ovcloud.warehouse.dto.resp.CertifyResp;
import com.hisense.ovcloud.warehouse.entity.julinkwarehouse.DeviceWarehouse;
import com.hisense.ovcloud.warehouse.entity.julinkwarehouse.OldDeviceWarehouse;
import com.hisense.ovcloud.warehouse.tool.SnowFlakeIDGenerator;
import com.mybatisflex.core.datasource.DataSourceKey;
import com.mybatisflex.core.row.Db;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.LongStream;

@Service
@Slf4j
public class DevicesService {
    private final WarehouseProperties warehouseProperties;
    private final ShardingJDBCExecutor shardingJDBCExecutor;

    private final LocalLFUCacheUtil<DeviceInfoInCache> deviceCache;


    private final AMQProducer amqProducer;

    private final WebClient webClient;

    private final WebClient domainServiceWebClient;

    public DevicesService(WarehouseProperties warehouseProperties, ShardingJDBCExecutor shardingJDBCExecutor, @Qualifier("deviceCache") LocalLFUCacheUtil<DeviceInfoInCache> deviceCache, ProductService productService, AMQProducer amqProducer, @Qualifier("base") WebClient webClient, @Qualifier("domainServiceWebClient") WebClient domainServiceWebClient) {
        this.warehouseProperties = warehouseProperties;
        this.shardingJDBCExecutor = shardingJDBCExecutor;
        this.deviceCache = deviceCache;
        this.amqProducer = amqProducer;
        this.webClient = webClient;
        this.domainServiceWebClient = domainServiceWebClient;
    }

    private long selectCountByBatchCode(String domainId, String batchCode) {
        String sql = Constants.JulinkWarehouse.SELECT_BY_BATCH_CODE
                .replaceAll(Constants.JulinkWarehouse.BATCH_CODE_TO_REPLACE, batchCode)
                .replaceAll(Constants.JulinkWarehouse.LOGIC_TABLE_NAME, Constants.JulinkWarehouse.BASE_LOGIC_TABLE_NAME + domainId);
        return shardingJDBCExecutor.directExecuteQueryCount(sql);
    }

    public Mono<BaseResponse> checkBatchCodeCount(Mono<BatchCodeCountReq> req) {
        return req.flatMap(r -> Mono.just(selectCountByBatchCode(r.getDomainId(), r.getBatchCode()))
                .zipWhen(count -> Mono.just(BaseResponse.builder().data(count).build()), (x, y) -> y));
    }

    @AllArgsConstructor
    @NoArgsConstructor
    @Builder
    @Data
    static
    class CleanCache {
        private String wifiIdList;
    }

    public Mono<BaseResponse> registerDevices(Mono<RegisterDevicesReq> req) {
        return req.flatMap(r -> Mono.just(r.getDevices().stream().map(d -> DeviceWarehouse.builder()
                        .domain(r.getDomain())
                        .device_id(d.getWifiId())
                        .batch_code(r.getBatchCode())
                        .public_key(d.getPublicKey())
                        .status(Constants.JulinkWarehouse.NOT_ACTIVATED)
                        .sub_domain(r.getSubDomain())
                        .sharding_key(MurMurHashUtils.hash(d.getWifiId()))
                        .build()).collect(Collectors.toList()))
                .map(totalList -> Pair.of(totalList, Lists.partition(totalList, warehouseProperties.getImportBatchSize())
                        .parallelStream().map(subList -> zipSql(r.getDomain(), subList)).toList()))
                .flatMap(pair -> {
                    shardingJDBCExecutor.directExecuteSQLBatch(pair.getValue());
                    return Mono.just(pair.getKey());
                })
                .zipWhen(totalList -> Mono.just(deviceCache.batchDelete(totalList.stream().map(DeviceWarehouse::getDevice_id).collect(Collectors.toList()))), (x, y) -> x)
                .zipWhen(totalList -> Mono.just(amqProducer.sendBlackListMsg(
                        JSONUtil.toJsonStr(
                                AMQMsg.builder().type(Constants.AMQConfig.MSG_TYPE_CLEAN_CACHE).data(
                                        JSONUtil.toJsonStr(CleanCache.builder().wifiIdList(
                                                totalList.stream().map(DeviceWarehouse::getDevice_id).collect(Collectors.joining(","))
                                        ).build())
                                ).build()

                        )
                )), (x, y) -> y)
                .zipWhen(result -> Mono.just(BaseResponse.builder()
                        .data(result)
                        .build()), (x, y) -> y)
        );
    }

    public Mono<BaseResponse> migrate(Mono<MigrateReq> req) {
        return req.flatMap(r -> {
            log.info("begin to migrate for domain: " + r.getDomain());
            return createSqlTable(Mono.just(CreateSqlReq.builder().domainId(r.getDomain()).build()))
                    .zipWhen(x -> {
                        long totalRowsNum = readOldDeviceWarehouseTablesRowsCount(r.getDomain());
                        log.info("total rows to migrate: {}", totalRowsNum);
                        long batchNum = totalRowsNum / r.getBatchSize();
                        long limit = r.getBatchSize();
                        List<OldDeviceWarehouse> oldList;
                        for (int i = 0; i <= batchNum; i++) {
                            long offset = i * r.getBatchSize();
                            if (i == batchNum) {
                                limit = totalRowsNum - r.getBatchSize() * i;
                            }
                            oldList = readOldDeviceWarehouseTables(r.getDomain(), limit, offset);
                            log.info("batch num {}, oldList size: {}. first one: {}", i, oldList.size(), oldList.getFirst().getDevice_id());
                            shardingJDBCExecutor.directExecuteOneSQL(zipSql(r.getDomain(), oldList.stream()
                                    .map(oldDeviceWarehouse -> DeviceWarehouse.builder()
                                            .domain(r.getDomain())
                                            .device_id(oldDeviceWarehouse.getDevice_id())
                                            .batch_code(oldDeviceWarehouse.getBatch_code())
                                            .public_key(oldDeviceWarehouse.getPublic_key())
                                            .status(oldDeviceWarehouse.getStatus())
                                            .sub_domain(oldDeviceWarehouse.getSub_domain())
                                            .create_time(Timestamp.valueOf(oldDeviceWarehouse.getCreate_time()))
                                            .update_time(Timestamp.valueOf(oldDeviceWarehouse.getModify_time()))
                                            .sharding_key(MurMurHashUtils.hash(Optional.ofNullable(oldDeviceWarehouse.getDevice_id()).orElse("NULL_VALUE")))
                                            .build()).toList()));
                        }
                        return Mono.just(BaseResponse.builder().data(String.format("succeed to migrate total %d rows to sharding tables for domain %s!", totalRowsNum, r.getDomain())).build());
                    }, (x, y) -> y);

        });
    }

    public Mono<BaseResponse> pressDataCreate(Mono<PressDataCreateReq> req) {
        return req.flatMap(r -> {
            log.debug("begin to create press data from {} to {}, the pubKey plain text is {}", r.getFeatureCode() + r.getWifiIdSuffix(), r.getFeatureCode() + (r.getWifiIdSuffix() + r.getInsertNum() - 1), "837cc1ef1f5eba96987d4d5b186734ba");
            return createSqlTable(Mono.just(CreateSqlReq.builder().domainId(r.getDomain()).build()))
                    .zipWhen(x -> {
                        shardingJDBCExecutor.directExecuteOneSQL(zipSql(r.getDomain(), LongStream.rangeClosed(r.getWifiIdSuffix(), r.getWifiIdSuffix() + r.getInsertNum() - 1).boxed().toList().stream()
                                .map(wifiIdSuffix -> DeviceWarehouse.builder()
                                        .domain(r.getDomain())
                                        .device_id(r.getFeatureCode() + wifiIdSuffix)
                                        .batch_code(r.getBatchCode())
                                        .public_key(r.getPubKeyAES())
                                        .status(1)
                                        .sub_domain("22")
                                        .create_time(Timestamp.valueOf(LocalDateTimeUtil.now()))
                                        .update_time(Timestamp.valueOf(LocalDateTimeUtil.now()))
                                        .sharding_key(MurMurHashUtils.hash(r.getFeatureCode() + wifiIdSuffix))
                                        .build()).toList()));
                        return Mono.just(BaseResponse.builder().data(String.format("succeed to create total %d rows to sharding tables for domain %s!", r.getInsertNum(), r.getDomain())).build());
                    }, (x, y) -> y);

        });
    }

    private List<OldDeviceWarehouse> readOldDeviceWarehouseTables(String domainId, long limit, long offset) {
        try {
            String sql = Constants.JulinkWarehouse.SELECT_ALL_OFFSET
                    .replaceAll(Constants.JulinkWarehouse.LIMIT_TO_REPLACE, String.valueOf(limit))
                    .replaceAll(Constants.JulinkWarehouse.OFFSET_TO_REPLACE, String.valueOf(offset))
                    .replaceAll(Constants.JulinkWarehouse.LOGIC_TABLE_NAME, domainId + Constants.JulinkWarehouse.ORIGIN_DEVICE_WAREHOUSE_TABLE_NAME_SUFFIX);
            DataSourceKey.use(Constants.DS.JULINK_WAREHOUSE);
            return Db.selectListBySql(sql).stream().map(row -> row.toEntity(OldDeviceWarehouse.class)).toList();
        } finally {
            DataSourceKey.clear();
        }
    }

    private long readOldDeviceWarehouseTablesRowsCount(String domainId) {
        try {
            String sql = Constants.JulinkWarehouse.SELECT_ALL_COUNT
                    .replaceAll(Constants.JulinkWarehouse.LOGIC_TABLE_NAME, domainId + Constants.JulinkWarehouse.ORIGIN_DEVICE_WAREHOUSE_TABLE_NAME_SUFFIX);
            DataSourceKey.use(Constants.DS.JULINK_WAREHOUSE);
            return Db.selectCount(sql);
        } finally {
            DataSourceKey.clear();
        }
    }

    public Mono<BaseResponse> createSqlTable(Mono<CreateSqlReq> req) {
        return req.flatMap(r -> {
                    List<String> sqlList = new ArrayList<>(List.of(
                            Constants.JulinkWarehouse.CREATE_JULINK_WAREHOUSE_DB.replaceAll(Constants.JulinkWarehouse.DOMAIN_ID_TO_REPLACE, r.getDomainId()),
                            Constants.JulinkWarehouse.CREATE_DEVICE_LICENSES_TABLE.replaceAll(Constants.JulinkWarehouse.DOMAIN_ID_TO_REPLACE, r.getDomainId()),
                            Constants.JulinkWarehouse.CREATE_FAILED_DEVICE_LICENSES_TABLE.replaceAll(Constants.JulinkWarehouse.DOMAIN_ID_TO_REPLACE, r.getDomainId()),
                            Constants.JulinkWarehouse.CREATE_DEVICE_PROPERTY_TABLE.replaceAll(Constants.JulinkWarehouse.DOMAIN_ID_TO_REPLACE, r.getDomainId()),
                            Constants.JulinkWarehouse.CREATE_DEVICE_ATOM_TABLE.replaceAll(Constants.JulinkWarehouse.DOMAIN_ID_TO_REPLACE, r.getDomainId())
                    ));
                    IntStream.rangeClosed(1, warehouseProperties.getShardingTableNum()).boxed().toList()
                            .forEach(tableNum -> sqlList.add(Constants.JulinkWarehouse.CREATE_DEVICE_TEMPLATE_TABLE
                                    .replaceAll(Constants.JulinkWarehouse.DOMAIN_ID_TO_REPLACE, r.getDomainId())
                                    .replaceAll(Constants.JulinkWarehouse.SERIAL_ID_TO_REPLACE, String.valueOf(tableNum)))
                            );

                    return Mono.just(sqlList);
                }).zipWhen(sqls -> Mono.just(shardingJDBCExecutor.directExecuteSQLBatch(sqls)), (x, y) -> y)
                .zipWhen(result -> Mono.just(BaseResponse.builder().build()), (x, y) -> y);
    }

    public Mono<BaseResponse> checkAndActivateDeviceThenCreateToken(Mono<ActivateDeviceReq> req) {
        return req.flatMap(r -> fetchDeviceInfoCache(r.getPhysicalDeviceId())
                .zipWhen(dtoInCache -> StringUtils.isEmpty(dtoInCache.getWifiId()) ?
                                Mono.just(Pair.of(dtoInCache, BaseResponse.builder().code(Constants.RESULT_FAILED).build())) :
                                Mono.just(Pair.of(dtoInCache, BaseResponse.builder().build())),
                        (x, y) -> y)
                .zipWhen(pair -> {
                    if (pair.getValue().getCode() != Constants.RESULT_FAILED) {
                        if (pair.getKey().getStatus() == Constants.JulinkWarehouse.NOT_ACTIVATED) {
                            return domainServiceWebClient
                                    .post()
                                    .uri(Constants.REST_SCHEMA_HTTP + warehouseProperties.getDomainServiceUrl() + "/save_wifi_dname")
                                    .body(Mono.just(SaveWifiDNameReq.builder().wifiId(r.getPhysicalDeviceId()).region(warehouseProperties.getRegion()).build()), SaveWifiDNameReq.class)
                                    .retrieve()
                                    .bodyToMono(Map.class)
                                    .doOnSuccess(res -> {
                                        updateStatus(pair.getKey().getDomain(), r.getPhysicalDeviceId());
                                        updateStatusInCache(r.getPhysicalDeviceId());
                                    })
                                    .doOnError(err -> log.error("failed to call save_wifi_dname for wifiId {}", r, err))
                                    .zipWhen(result -> Mono.just(result.containsKey("resultCode") && Optional.ofNullable(result.get("resultCode").toString()).orElse("1").equals("0")), (x, y) -> y)
                                    .flatMap(success -> success ? Mono.just(Constants.RESULT_SUCCESS) : Mono.just(Constants.RESULT_FAILED));
                        }
                        return Mono.just(Constants.RESULT_SUCCESS);
                    } else {
                        return Mono.just(Constants.RESULT_FAILED);
                    }
                }, (x, y) -> y)
                .zipWhen(lastStepResult -> {
                    if (lastStepResult == Constants.RESULT_SUCCESS) {
                        if (r.isNeedToken()) {
                            return webClient
                                    .post()
                                    .uri(Constants.REST_SCHEMA_HTTP + warehouseProperties.getPsUrl() + "/ac-julink-ps/certify")
                                    .body(Mono.just(CertifyReq.builder().wifiId(r.getPhysicalDeviceId()).build()), CertifyReq.class)
                                    .retrieve()
                                    .bodyToMono(BaseResponse.class)
                                    .doOnSuccess(res -> log.info("success to call certify for wifiId {}", r.getPhysicalDeviceId()))
                                    .doOnError(err -> log.error("failed to call certify for wifiId {}", r.getPhysicalDeviceId(), err))
                                    .zipWhen(result -> result.getCode() == Constants.RESULT_SUCCESS ?
                                            Mono.just(BaseResponse.builder().data(ActivateResp.builder().activated(true).token(JSONUtil.toBean(JSONUtil.toJsonStr(result.getData()), CertifyResp.class).getToken()).build()).build())
                                            : Mono.just(BaseResponse.builder().code(Constants.RESULT_FAILED).data(ActivateResp.builder().build()).build()), (x, y) -> y);
                        } else {
                            return Mono.just(BaseResponse.builder().data(ActivateResp.builder().activated(true).build()).build());
                        }

                    } else {
                        return Mono.just(BaseResponse.builder().code(Constants.RESULT_FAILED).data(ActivateResp.builder().build()).build());
                    }
                }, (x, y) -> y));
    }

    // former authBatch
    public Mono<BaseResponse> authBatch(Mono<AuthBatchReq> req) {
        return req.flatMap(r -> fetchDeviceInfoCache(r.getWifiId())
                .zipWhen(result -> StringUtils.isEmpty(result.getWifiId()) ?
                                Mono.just(BaseResponse.builder().code(Constants.RESULT_FAILED).data(AuthBatchResp.builder().flag(false).build()).build()) :
                                Mono.just(BaseResponse.builder().data(AuthBatchResp.builder().flag(true).build()).build()),
                        (x, y) -> y));
    }

    public Mono<AuthBatchResp> authBatchOrigin(Mono<AuthBatchReq> req) {
        return req.flatMap(r -> fetchDeviceInfoCache(r.getWifiId())
                .zipWhen(result -> StringUtils.isEmpty(result.getWifiId()) ?
                        Mono.just(AuthBatchResp.builder().flag(false).build()) : Mono.just(AuthBatchResp.builder().flag(true).build()),
                        (x, y) -> y));
    }

    public Mono<AuthTokenAndBatchResp> authTokenAndBatch(Mono<AuthTokenAndBatchReq> req) {
        return req.flatMap(r -> fetchDeviceInfoCache(r.getWifiId())
                .zipWhen(result -> {
                            if(StringUtils.isEmpty(result.getWifiId())){
                                log.error("{} auth token and batch faild,reason: {}", r.getWifiId(), "auth device failed, check warehouse for details");
                                return Mono.just(AuthTokenAndBatchResp.builder().batchFlag(Constants.RESULT_FAILED).sign("").build());
                            }else{
                                return fetchDeviceInfoCache(r.getWifiId())
                                        .zipWhen(deviceInfoInCache -> {
                                            if(StringUtils.isNotEmpty(deviceInfoInCache.getPlainPubKey())){
                                                String sign = DigestUtils.sha256Hex(r.getData()+deviceInfoInCache.getPlainPubKey());
                                                log.info("sign and batch all success: wifi_id is [{}],sign is [{}}]", r.getWifiId(), sign);
                                                return Mono.just(AuthTokenAndBatchResp.builder().batchFlag(Constants.RESULT_SUCCESS).sign(sign).build());
                                            }else{
                                                log.error("{} auth token and batch faild,reason: {}", r.getWifiId(), "pubkey not found!");
                                                return Mono.just(AuthTokenAndBatchResp.builder().batchFlag(Constants.RESULT_FAILED).sign("").build());
                                            }
                                        }, (x, y) -> y);
                            }
                        }, (x, y) -> y));
    }


    public Mono<BaseResponse> getPublicKey(Mono<ActivateDeviceReq> req) {
        return req.flatMap(r -> fetchDeviceInfoCache(r.getPhysicalDeviceId())
                .zipWhen(result -> {
                    if(StringUtils.isEmpty(result.getWifiId())){
                        return Mono.just(BaseResponse.builder().code(Constants.RESULT_FAILED).build());
                    }else{
                        if(!r.isHencDevice()){
                            return Mono.just(BaseResponse.builder().data(result.getPlainPubKey()).build());
                        }else{
                            log.info("path2");
                            return Mono.just(BaseResponse.builder().data(new HiAESUtil().encWithPreKey(result.getPlainPubKey(), result.getPlainPubKey())).build());
                        }
                    }},(x, y) -> y));
    }

    public Mono<String> getPublicKeyInner(Mono<String> req) {
        return req.flatMap(wifiId -> fetchDeviceInfoCache(wifiId)
                .zipWhen(result -> StringUtils.isEmpty(result.getWifiId()) ?
                                Mono.just("") :
                                Mono.just(result.getPlainPubKey()),
                        (x, y) -> y));
    }

    private void updateStatus(String domainId, String wifiId) {
        String sql = Constants.JulinkWarehouse.UPDATE_STATUS
                .replaceAll(Constants.JulinkWarehouse.WIFI_ID_TO_REPLACE, wifiId)
                .replaceAll(Constants.JulinkWarehouse.STATUS_TO_REPLACE, String.valueOf(Constants.JulinkWarehouse.ACTIVATED))
                .replaceAll(Constants.JulinkWarehouse.SHARDING_KEY_TO_REPLACE, String.valueOf(MurMurHashUtils.hash(wifiId)))
                .replaceAll(Constants.JulinkWarehouse.LOGIC_TABLE_NAME, Constants.JulinkWarehouse.BASE_LOGIC_TABLE_NAME + domainId);
        shardingJDBCExecutor.directExecuteOneSQL(sql);
    }

    public int updatePemVersion(String domainId, String wifiId, int pemVersion) {
        log.info("update pem version: wifiId is [{}], pemVersion is [{}]", wifiId, pemVersion);
        String sql = Constants.JulinkWarehouse.UPDATE_PEM_VERSION
                .replaceAll(Constants.JulinkWarehouse.WIFI_ID_TO_REPLACE, wifiId)
                .replaceAll(Constants.JulinkWarehouse.PEM_VERSION_TO_REPLACE, String.valueOf(pemVersion))
                .replaceAll(Constants.JulinkWarehouse.SHARDING_KEY_TO_REPLACE, String.valueOf(MurMurHashUtils.hash(wifiId)))
                .replaceAll(Constants.JulinkWarehouse.LOGIC_TABLE_NAME, Constants.JulinkWarehouse.BASE_LOGIC_TABLE_NAME + domainId);
        shardingJDBCExecutor.directExecuteOneSQL(sql);

        return 0;
    }

    private List<DeviceWarehouse> selectByWifiId(String wifiId) {
        return ReadShardingTablesConfig.domainsList.parallelStream().map(domainId -> {
            String sql = Constants.JulinkWarehouse.SELECT_BY_WIFI_ID
                    .replaceAll(Constants.JulinkWarehouse.WIFI_ID_TO_REPLACE, wifiId)
                    .replaceAll(Constants.JulinkWarehouse.SHARDING_KEY_TO_REPLACE, String.valueOf(MurMurHashUtils.hash(wifiId)))
                    .replaceAll(Constants.JulinkWarehouse.LOGIC_TABLE_NAME, Constants.JulinkWarehouse.BASE_LOGIC_TABLE_NAME + domainId);
            return shardingJDBCExecutor.directExecuteQueryAndMapToBeanV2(sql, DeviceWarehouse.class);
        }).flatMap(List::stream).collect(Collectors.toList());
    }

    public Mono<DeviceInfoInCache> fetchDeviceInfoCache(String wifiId) {
        return Mono.just(Optional.ofNullable(deviceCache.getFromCache(wifiId)).orElse(DeviceInfoInCache.builder().build())).zipWhen(dtoInCache -> {
            if (StringUtils.isNotEmpty(dtoInCache.getWifiId())) {
                return Mono.just(dtoInCache);
            } else {
                return Mono.just(selectByWifiId(wifiId)).zipWhen(deviceWareResult -> {
                    if (deviceWareResult != null && !deviceWareResult.isEmpty()) {
                        DeviceWarehouse deviceWarehouse = deviceWareResult.getFirst();
                        String pubkey = deviceWarehouse.getPublic_key();
                        if(pubkey.startsWith(Constants.AESKEY_PREFIX)){
                            pubkey = AES256CbcUtil.decrypt256CBCWithoutIv(warehouseProperties.getImportAES256CBCKey(),
                                    pubkey.replaceAll(Constants.AESKEY_PREFIX, ""));
                        }
                        log.debug("pubkey is {} for wifiId {}", pubkey, deviceWareResult.getFirst().getDevice_id());
                        return Mono.just(DeviceInfoInCache.builder()
                                .wifiId(wifiId)
                                .domain(deviceWareResult.getFirst().getDomain())
                                .subDomain(deviceWareResult.getFirst().getSub_domain())
                                .plainPubKey(pubkey)
                                .status(deviceWareResult.getFirst().getStatus())
                                .pemVersion(deviceWareResult.getFirst().getPem_version())
                                .build()).zipWhen(newInfo -> {
                            deviceCache.saveToCache(wifiId, newInfo);
                            return Mono.just(newInfo);
                        }, (x, y) -> y);
                    }
                    return Mono.just(DeviceInfoInCache.builder().build());
                }, (x, y) -> y);
            }
        }, (x, y) -> y);
    }

    private void updateStatusInCache(String wifiId){
        DeviceInfoInCache deviceInfoInCacheOld = deviceCache.getFromCache(wifiId);
        if(null != deviceInfoInCacheOld){
            deviceInfoInCacheOld.setStatus(Constants.JulinkWarehouse.ACTIVATED);
            deviceCache.saveToCache(wifiId, deviceInfoInCacheOld);
        }
    }

    public boolean updatePemVersionInCache(String wifiId, int pemVersion){
        DeviceInfoInCache deviceInfoInCacheOld = deviceCache.getFromCache(wifiId);
        if(null != deviceInfoInCacheOld){
            deviceInfoInCacheOld.setPemVersion(pemVersion);
            deviceCache.saveToCache(wifiId, deviceInfoInCacheOld);
        }

        return true;
    }

    private String zipSql(String domainId, List<DeviceWarehouse> req) {
        return Constants.JulinkWarehouse.INSERT_PREFIX
                .replaceAll(Constants.JulinkWarehouse.LOGIC_TABLE_NAME, Constants.JulinkWarehouse.BASE_LOGIC_TABLE_NAME + domainId)
                + req.stream().map(info -> String.format("('%s', '%s', '%s', '%s', '%s', %d, '%s', %d, UTC_TIMESTAMP(), UTC_TIMESTAMP())",
                SnowFlakeIDGenerator.generate(),
                domainId,
                info.getSub_domain(),
                info.getDevice_id(),
                !info.getPublic_key().startsWith(Constants.AESKEY_PREFIX) ? String.format("%s%s", Constants.AESKEY_PREFIX,
                        AES256CbcUtil.encrypt256CBCWithoutIv(warehouseProperties.getImportAES256CBCKey(), info.getPublic_key().getBytes(StandardCharsets.UTF_8))) : info.getPublic_key(),
                info.getStatus(),
                info.getBatch_code(),
                MurMurHashUtils.hash(info.getDevice_id())
        )).collect(Collectors.joining(",")) + Constants.JulinkWarehouse.INSERT_SUFFIX;
    }

    public static void main(String[] args) {
        System.out.println(new HiAESUtil().encWithPreKey("550c73881ee9cd3290490488e0a8c3ea", "550c73881ee9cd3290490488e0a8c3ea"));
    }
}
